package org.raymond.iworks.study.mq.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.raymond.iworks.study.mq.kafka.common.Foo2;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
@Component
public class DemoConsumer {
    private final TaskExecutor exec = new SimpleAsyncTaskExecutor();
    @KafkaListener(id = "group_demo1", topics = "demo1")
    public void listen1(String input) {
        log.info("Received1: " + input);
    }
    @KafkaListener(id = "group_demo2", topics = "demo2")
    public void listen2(String input) {
        log.info("Received2: " + input);
    }
    /*
    @KafkaListener(id = "group_demo3", topics = "demo3")
    public void listen3(LogModel logModel) {
        log.info("Received3: " + logModel);
    }*/
    @KafkaListener(id = "group_demo4", topics = "demo4")
    public void listen4(Foo2 foo) {
        log.info("Received4: {}", foo);
    }
    @KafkaListener(id = "group_demo5", topics = "demo5")
    public void listen5(Foo2 foo) {
        log.info("Received5: {}", foo);
        if (foo.getFoo().startsWith("fail")) {
            throw new RuntimeException("failed");
        }
    }
    @KafkaListener(id = "group_demo5_dlt", topics = "demo5.DLT")
    public void listen5dlt(byte[] in) {
        log.info("Received5 dlt: {}", new String(in));
    }
    @KafkaListener(id = "group_demo6", topics = "demo6")
    public void listen6(Foo2 foo) {
        log.info("Received6: {}", foo);
    }
}
